package com.atguigu.realtime.app.dwd.db;


import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;


/**
 * com.atguigu.realtime.app.dwd.db.Dwd_12_DwdToolCouponOrder
 * 工具域优惠券使用(下单)事务事实表（练习）
 *	读取优惠券领用表数据，筛选优惠券下单数据，写入 Kafka 优惠券下单主题。
 *  用户使用优惠券下单时，优惠券领用表的 using_time 字段会更新为下单时间，
 *  同时 coupon_status 字段会由 1401 更改为 1402，因此优惠券下单数据应满足三个条件：
 *  ① 操作类型为 update；
 *  ② 当前 coupon_status 字段的值为 1402；
 *  ③ 修改了 coupon_status 字段。
 */
public class Dwd_12_DwdToolCouponOrder extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_12_DwdToolCouponOrder().init(
                3012,2,
                "Dwd_12_DwdToolCouponOrder"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        // 1. 读取ods_db
        readOdsDb(tEnv, "Dwd_11_DwdToolCouponGet");


        // TODO 2. 读取优惠券领用表数据，筛选优惠券使用（支付）数据
        Table result = tEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['coupon_id'] coupon_id,\n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "date_format(data['using_time'],'yyyy-MM-dd') date_id,\n" +
                "data['using_time'] using_time,\n" +
                "ts\n" +
                "from ods_db\n" +
                "where `database`='gmall2022' \n" +
                "and `table` = 'coupon_use'\n" +
                "and `type` = 'update'\n" +
                "and data['coupon_status'] = '1402'\n" +
                "and `old`['coupon_status'] = '1401'");


        tEnv.executeSql("create table dwd_tool_coupon_order(\n" +
                "id string,\n" +
                "coupon_id string,\n" +
                "user_id string,\n" +
                "order_id string,\n" +
                "date_id string,\n" +
                "order_time string,\n" +
                "ts bigint \n" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TOOL_COUPON_ORDER));

        result.executeInsert("dwd_tool_coupon_order");
    }
}

/*

 */